-- Copyright (c) 2024 Huawei Technologies Co., Ltd.
-- openUBMC is licensed under Mulan PSL v2.
-- You can use this software according to the terms and conditions of the Mulan PSL v2.
-- You may obtain a copy of Mulan PSL v2 at: http://license.coscl.org.cn/MulanPSL2
-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
-- See the Mulan PSL v2 for more details.

local lu = require 'luaunit'
local cmn = require 'common'
local c_device_service = require 'device.device_service'
local c_pcie_device = require 'device.class.pcie_device'
local c_ocp_card = require 'device.class.ocp_card'
local c_pcie_card = require 'device.class.pcie_card'
local c_pcie_card_flash_checker = require 'device.class.pcie_card_flash_checker'
local utils = require 'mc.utils'
local client = require 'pcie_device.client'
local log = require 'mc.logging'
local c_pcie_addr_info = require 'biz_topo.class.pcie_addr_info'
local event = require 'infrastructure.event'
local cjson = require 'cjson'
local def = require 'biz_topo.def'
local context = require 'mc.context'
local c_pcie_slot = require 'device.class.pcie_slot'
TestDeviceService = {}

local DATA_PATH<const> = './test_data/'
local PCIECARD_INTERFACE <const> = 'bmc.kepler.Systems.PCIeDevices.PCIeCard'

function TestDeviceService:test_method_card_info_dump()
    client.GetPackageObjects = function()
        return {}
    end
    local pcie_card_list = {
        [1] = {
            get_prop = function(_, prop)
                local props = {
                    SlotID = 3,
                    LaneOwner = 0,
                    VendorID = 4096,
                    DeviceID = 4322,
                    SubVendorID = 4096,
                    SubDeviceID = 16400,
                    Description = 'MegaRAID 9560-8i',
                    BoardID = 0,
                    PcbVersion = 'N/A',
                    PartNumber = '06030622',
                    Manufacturer = 'Broadcom',
                    Position = 'PCIeRiser1',
                    Name = '9560-8i'
                }
                return props[prop]
            end
        }
    }
    local ocp_card_list = {
        [1] = {
            get_prop = function(_, prop)
                local props = {
                    SlotID = 3,
                    LaneOwner = 0,
                    VendorID = 4097,
                    DeviceID = 4323,
                    SubVendorID = 4097,
                    SubDeviceID = 16401,
                    Description = 'MT2892 Family [ConnectX-6 Dx]',
                    BoardID = 65535,
                    PcbVersion = '.A',
                    PartNumber = '06310299',
                    Manufacturer = 'Mellanox',
                    Position = 'EXU',
                    Name = 'MCX623436MN-CDAB'
                }
                return props[prop]
            end
        }
    }
    local cpu_table = {
        {LogicalId = 2, PhysicalId = 3},
        {LogicalId = 0, PhysicalId = 1}
    }
    local c_ForeachCPUObjects = client.ForeachCPUObjects
    client.ForeachCPUObjects = function(self, cb)
        for _, obj in pairs(cpu_table) do
            cb(obj)
        end
    end

    c_device_service.pcie_device_list = {}
    c_device_service.pcie_card_list = pcie_card_list
    c_device_service.ocp_card_list = ocp_card_list
    c_device_service:method_card_info_dump(nil, utils.realpath(DATA_PATH))
    client.ForeachCPUObjects = c_ForeachCPUObjects
    local test_file_path = DATA_PATH .. 'card_info'
    local file = io.open(test_file_path, 'r')
    local str = file:read("a")
    file:close()
    lu.assertEquals(str:sub(1, 14), 'Pcie Card Info')
    lu.assertEquals(str:sub(798, 810), 'OCP Card Info')
    os.execute('rm -f ' .. test_file_path)
end

function TestDeviceService:test_on_add_ocp_card()
    c_device_service.ocp_card_list = {}
    c_device_service:on_add_object('OCPCard', {
        PcbID = 1,
        PcbVersion = ''
    }, 'position')
    lu.assertEquals(#c_device_service.ocp_card_list, 1)

    local pcie_device_obj = c_pcie_device.new({
        DeviceName = "pcie-device",
        DeviceType = 8,
        SocketID = 1,
        DevBus = 150,
        DevDevice = 0,
        DevFunction = 0,
        BaseClassCode = 0,
        SubClassCode = 0,
        ProgrammingInterface = 0,
        MultihostPresence = 1,
        path = '/bmc/kepler/Systems/1',
        property_changed = {
            on = function()
            end
        }
    }, "position", {})

    cmn.skynet.fork_loop = function(_, cb)
        cb()
    end
    c_device_service:register_get_pcie_info_task(pcie_device_obj, "position")

    local position = "position1"
    local mdb_obj = {
        DeviceName = "ocp1",
        DeviceType = 8,
        SocketID = 1,
        DevBus = 150,
        DevDevice = 0,
        DevFunction = 0,
        BaseClassCode = 0,
        SubClassCode = 0,
        ProgrammingInterface = 0,
        MultihostPresence = 1,
        path = '/bmc/kepler/Systems/1',
        PcbID = 1,
        PcbVersion = ''
    }
    local bus = {}
    local reset_local_db = {}
    local ocp_obj = c_ocp_card.new(mdb_obj, position, bus, reset_local_db)
    ocp_obj.get_pcie_lang_info = function()
    end
    ocp_obj.sync_info_to_card = function()
    end
    ocp_obj.update_pcie_vpd_info = function()
    end
    local sleep_bak = cmn.skynet.sleep
    cmn.skynet.sleep = function()
        log:raise("break")
    end
    local ok, res = pcall(function()
        c_device_service.pcie_and_ocp_task(ocp_obj, {ocp_obj}, "position1")
    end)
    lu.assertEquals(ok, false)
    lu.assertNotEquals(res, nil)
    cmn.skynet.sleep = sleep_bak
end

function TestDeviceService:test_sync_eth_device_bdf_info()
    local addr_obj = {
        get_prop = function() return true end
    }
    local biz_topo = {
        pfid_map = {[1] = {[0] = 0, [1] = 4}}
    }
    c_device_service.service_manager = {
        get_service = function() return {
            biz_topo = biz_topo,
            method_get_target_conn_by_addr_obj = function() return true end
        } end,
    }
    obj_list = {
        ethobj1 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 0, sync_ready = true,
            mds_obj = { name = "port1", PortId = 0, Device = 0,
                Function = 0, PortFunctionId = 0 }
        },
        ethobj2 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 0, sync_ready = true,
            mds_obj = { name = "port2", PortId = 1, Device = 0,
                Function = 0, PortFunctionId = 0 }
        },
        ethobj3 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 1, sync_ready = true,
            mds_obj = { name = "port3", PortId = 2, Device = 0,
                Function = 0, PortFunctionId = 0 }
        },
        ethobj4 = {
            position = '0101', sync_bdf_info = function() end,
            bcu_idx = 1, die_id = 1, sync_ready = true,
            mds_obj = { name = "port4", PortId = 3, Device = 0,
                Function = 0, PortFunctionId = 0 }
        }
    }
    cmn.skynet.fork = function(cb) cb() end
    c_device_service:sync_eth_device_bdf_info(obj_list, addr_obj)
    lu.assertEquals(obj_list.ethobj1.mds_obj.PortFunctionId, 0)
    lu.assertEquals(obj_list.ethobj2.mds_obj.PortFunctionId, 1)
    lu.assertEquals(obj_list.ethobj3.mds_obj.PortFunctionId, 4)
    lu.assertEquals(obj_list.ethobj4.mds_obj.PortFunctionId, 5)
    lu.assertEquals(obj_list.ethobj1.mds_obj.Function, 0)
    lu.assertEquals(obj_list.ethobj2.mds_obj.Function, 1)
    lu.assertEquals(obj_list.ethobj3.mds_obj.Function, 0)
    lu.assertEquals(obj_list.ethobj4.mds_obj.Function, 1)
end

function TestDeviceService:test_sync_pcie_addr_info_to_device()
    local pcie_addr_info_obj = c_pcie_addr_info.new({
        ComponentType = 83,
        SlotID = 1,
        path = '/bmc/kepler/Systems/1'
    }, "position", {})

    c_device_service.service_manager = {
        get_service = function()
            return {
                biz_topo = {
                    get_objs = function()
                        return {
                            pcie_addr_info_obj
                        }
                    end
                }
            }
        end
    }
    cmn.skynet.fork = function(cb)
        cb()
    end
    local pcie_device_obj = c_pcie_device.new({
        DeviceName = "pcie-device",
        DeviceType = 83,
        SlotID = 1,
        SocketID = 1,
        DevBus = 150,
        DevDevice = 0,
        DevFunction = 0,
        BaseClassCode = 0,
        SubClassCode = 0,
        ProgrammingInterface = 0,
        MultihostPresence = 1,
        path = '/bmc/kepler/Systems/1',
        property_changed = {
            on = function()
            end
        }
    }, "position", {})

    pcie_device_obj.sync_pcie_addr_info = function()
    end
    c_device_service.update_pcie_type = function()
    end
    local sleep_bak = cmn.skynet.sleep
    cmn.skynet.sleep = function()
        log:raise("break")
    end
    local ok, res = pcall(function()
        c_device_service:sync_pcie_addr_info_to_device(pcie_device_obj)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(res, nil)
    cmn.skynet.sleep = sleep_bak
end

function TestDeviceService:test_on_add_flash_checker()
    c_device_service.flash_checker_list = {}
    c_device_service:on_add_object('PCIeCardFlashChecker', {
        prop = ''
    }, 'position')
    lu.assertEquals(#c_device_service.flash_checker_list, 1)

    c_device_service:on_delete_object('PCIeCardFlashChecker', {
        prop = ''
    }, 'position')
    lu.assertEquals(#c_device_service.flash_checker_list, 0)
end

function TestDeviceService:test_update_flash_checker()
    local position = 'position'
    local mdb_obj = {
        PcbID = 1,
        PcbVersion = '',
        SerialNumber = '',
        PartNumber = '',
        CardForServers = {},
        LinkSpeed = '',
        LinkSpeedCapability = '',
        LinkWidth = '',
        LinkWidthAbility = '',
        Protocol = '',
        MaxFrameLen = '',
        RefChip = '',
        DeviceName = "pcie-card",
        DeviceType = 8,
        VendorID = 0x1111,
        DeviceID = 0x2222,
        SlotID = 1,
        [PCIECARD_INTERFACE] = {
        }
    }
    local bus = {}
    local reset_local_db = {}
    local pcie_card_obj = c_pcie_card.new(mdb_obj, position, bus, reset_local_db)

    local pcie_card_flash_checker_obj = c_pcie_card_flash_checker.new({
        ["FaultStatus"] = 0,
        ["PCIeQuadrupleList"] = {{'1111_2222','card-des'}}
    }, "position", {})

    local flash_checker_list = {pcie_card_flash_checker_obj}

    c_device_service.update_flash_checker(pcie_card_obj, "position", flash_checker_list, {})
    lu.assertEquals(pcie_card_flash_checker_obj.mds_obj.FaultStatus, 1)

    pcie_card_obj:set_prop('VendorID', 123)
    c_device_service.update_flash_checker(pcie_card_obj, "position", flash_checker_list, {})
    lu.assertEquals(pcie_card_flash_checker_obj.mds_obj.FaultStatus, 0)

    flash_checker_list = {}
    c_device_service.update_flash_checker(pcie_card_obj, "position", flash_checker_list, {})
    lu.assertEquals(pcie_card_flash_checker_obj.mds_obj.FaultStatus, 0)
end

function TestDeviceService:test_update_init_abnormal_event()
    local get_latest_alarm_list_bak = event.get_latest_alarm_list
    event.get_latest_alarm_list = function()
        return {{
            EventCode = '0x080000A5',
            MessageArgs = cjson.encode({'', 1, 'card description', "Flash fault"}),
        }}
    end

    local bus = {}
    local status = 1
    local slot = 0
    local card_des = 'card description'
    local device_name = 'card device_name'
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)

    status = 0
    slot = 1
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)

    local generate_event_bak = cmn.generate_event
    cmn.generate_event = function()
    end
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)
    cmn.generate_event = function()
        log:raise('error')
    end
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)

    status = 1
    slot = 2
    c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)
    cmn.generate_event = function()
    end
    local ok, res = pcall(function()
        c_device_service.update_init_abnormal_event(bus, status, slot, card_des, device_name)
    end)
    lu.assertEquals(ok, true)
    lu.assertEquals(res, nil)
    
    event.get_latest_alarm_list = get_latest_alarm_list_bak
    cmn.generate_event = generate_event_bak
end

function TestDeviceService:test_clear_flash_checker_event()
    local sleep_bak = cmn.skynet.sleep
    local update_init_abnormal_event_bak = c_device_service.update_init_abnormal_event
    local get_latest_alarm_list_bak = c_device_service.get_latest_alarm_list
    cmn.skynet.sleep = function()
    end
    c_device_service.update_init_abnormal_event = function()
    end
    c_device_service.get_latest_alarm_list = function()
        return {[1] = true}
    end

    local ok, _ = pcall(function()
        c_device_service.clear_flash_checker_event({}, 0)
    end)
    lu.assertEquals(ok, true)
    cmn.skynet.sleep = sleep_bak
    c_device_service.update_init_abnormal_event = update_init_abnormal_event_bak
    c_device_service.get_latest_alarm_list = get_latest_alarm_list_bak
end

function TestDeviceService:test_on_add_pcie_slot()
    c_device_service.pcie_slot_list = {}
    local pcie_slot_obj = {
        SlotId = 1,
        SupportedComponentTypes = {8},
        PowerState = "On",
        ReadyToRemove = 0
    }
    c_device_service:on_add_object("PCIeSlot", pcie_slot_obj, "position")
    lu.assertEquals(#c_device_service.pcie_slot_list, 1)
    lu.assertEquals(c_device_service.pcie_slot_list[1].mds_obj.SlotId, 1)
    lu.assertEquals(c_device_service.pcie_slot_list[1].mds_obj.SupportedComponentTypes[1], 8)
    lu.assertEquals(c_device_service.pcie_slot_list[1].mds_obj.PowerState, "On")
end

function TestDeviceService:test_set_pcie_slot_power_off()
    local ctx = {}
    ctx.get_initiator = function()
        return 'test'
    end
    local obj = {
        SlotId = 2,
        position = "position"
    }
    local res = pcall(c_device_service.set_pcie_slot_power_state, c_device_service, obj, ctx, "Off")
    lu.assertEquals(res, false)
    local pcie_slot_obj = {
        SlotId = 2,
        SupportedComponentTypes = {8},
        PowerState = "On",
        ReadyToRemove = 0
    }
    c_device_service:on_add_object("PCIeSlot", pcie_slot_obj, "position")
    res = pcall(c_device_service.set_pcie_slot_power_state, c_device_service, obj, ctx, "Off")
    lu.assertEquals(res, true)
    lu.assertEquals(c_device_service.pcie_slot_list[1].mds_obj.ReadyToRemove, 0)
    lu.assertEquals(c_device_service.pcie_slot_list[2].mds_obj.ReadyToRemove, 1)

    local pcie_slot_obj_1 = {
        SlotId = 2,
        SupportedComponentTypes = {1, 57},
        PowerState = "Off",
        PowerOnSwitch = 1
    }
    c_device_service:on_add_object("PCIeSlot", pcie_slot_obj_1, "position")
    res = pcall(c_device_service.set_pcie_slot_power_state, c_device_service, obj, ctx, "On")
    lu.assertEquals(res, true)
    lu.assertEquals(c_device_service.pcie_slot_list[3].mds_obj.PowerOnSwitch, 1)

    c_pcie_slot:set_power_state("On", {57})
end

function TestDeviceService:test_load_function_record()
    local disk_list = {}

    local pcie_card_list = {
        {
            DeviceType = "PCIeCard",
            SlotId = 2,
            SegmentNumber = 0,
            BusNumber = 0x40,
            DeviceNumber = 0,
            FunctionNumber = 0,
            LogicProcessorId = 0,
            SystemId = 1
        },
        {
            DeviceType = "PCIeCard",
            SlotId = 2,
            SegmentNumber = 0,
            BusNumber = 0x1c,
            DeviceNumber = 0,
            FunctionNumber = 0,
            LogicProcessorId = 1,
            SystemId = 1
        },
        {
            DeviceType = "PCIeCard",
            SlotId = 1,
            SegmentNumber = 0,
            BusNumber = 0x16,
            DeviceNumber = 0,
            FunctionNumber = 0,
            LogicProcessorId = 0,
            SystemId = 1
        }
    }

    local ocp_card = {}

    local get_table = {
        [def.device_type.DISK] = {
            all = function(...)
                return {}
            end
        },
        [def.device_type.PCIE_CARD] = {
            all = function(...)
                return pcie_card_list
            end,
        },
        [def.device_type.OCP_CARD] = {
            all = function(...)
                return {}
            end
        }
    }

    local where_list = {
        where = function(_self, device_type)
            return get_table[device_type]
        end
    }
    
    local reset_local_db = {
        select = function(...)
            return where_list
        end
    }

    reset_local_db.PCIeFunctionLoadInfo = {
        DeviceType = {
            eq = function(_self, device_type)
                return device_type
            end
        }
    }

    c_device_service.bus = {}
    c_device_service.reset_local_db = reset_local_db
    c_device_service:main()

    local table_num = 0

    for _, lists in pairs(c_device_service.pcie_function_list) do
        for _, list in pairs(lists) do
            for _, _ in pairs(list) do
                table_num = table_num + 1
            end
        end
    end
    lu.assertEquals(table_num, #disk_list + #pcie_card_list + #ocp_card)
end

function TestDeviceService:test_set_pcie_bdf()
    local reset_local_db = {
        PCIeFunctionLoadInfo = function(record)
            record.save = function()
            end
            return record
        end
    }

    local card_infos = {
        {
            SlotId = 1,
            SegmentNumber = 0,
            LogicProcessorId = 0,
            BusNumber = 0x1c,
            DeviceNumber = 0,
            FunctionNumber = 0
        },
        {
            SlotId = 2,
            SegmentNumber = 0,
            LogicProcessorId = 0,
            BusNumber = 0x40,
            DeviceNumber = 0,
            FunctionNumber = 0
        }
    }

    local biz_topo_service = {
        method_get_pcie_addr_by_pos = function(...)
            return {
                get_prop = function()
                    return 1
                end
            }
        end
    }

    local service_manager = {
        get_service = function(...)
            return biz_topo_service
        end
    }

    c_device_service.service_manager = service_manager

    local pcie_device_mds = {
        path = "/bmc/kepler/Systems/1/PCIeDevices/PCIeDevice_1_01010101",
        property_changed = {
            on = function(...) end
        }
    }

    c_device_service.pcie_device_list = {
        c_pcie_device.new(pcie_device_mds, "01010101", {})
    }
    
    local saved = c_device_service.uptree_pcie_function_of_pcie_device
    c_device_service.pcie_function_list = {}
    c_device_service.reset_local_db = reset_local_db
    c_device_service:on_set_pcie_bdf(1, "PCIeCard", card_infos)

    local table_num = 0
    for _, lists in pairs(c_device_service.pcie_function_list) do
        for _, list in pairs(lists) do
            for _, _ in pairs(list) do
                table_num = table_num + 1
            end
        end
    end
    lu.assertEquals(table_num, #card_infos)
    c_device_service.uptree_pcie_function_of_pcie_device = saved
end

function TestDeviceService:test_get_vdss()
    local pcie_device_mds_pcie = {
        path = "/bmc/kepler/Systems/1/PCIeDevices/PCIeDevice_1_01010101",
        property_changed = {
            on = function(...) end
        },
        DeviceType = 8
    }

    local pcie_device_mds_ocp = {
        path = "/bmc/kepler/Systems/1/PCIeDevices/PCIeDevice_1_01010102",
        property_changed = {
            on = function(...) end
        },
        DeviceType = 83
    }

    local pcie_card_device = c_pcie_device.new(pcie_device_mds_pcie, "01010101", {})
    local ocp_card_device = c_pcie_device.new(pcie_device_mds_ocp, "01010102", {})
    
    c_device_service.pcie_card_list = {
        {
            position = "01010101",
            mds = {
                VendorID = 1, DeviceID = 1, SubDeviceID = 1, SubVendorID = 1
            }
        }
    }

    c_device_service.ocp_card_list = {
        {
            position = "01010102",
            mds = {
                VendorID = 2, DeviceID = 2, SubDeviceID = 2, SubVendorID = 2
            }
        }
    }

    local vdss_pcie = c_device_service:get_pcie_function_vdss(pcie_card_device)
    lu.assertEquals(vdss_pcie, {1,1,1,1})

    local vdss_ocp = c_device_service:get_pcie_function_vdss(ocp_card_device)
    lu.assertEquals(vdss_ocp, {2,2,2,2})
end
